package com.joizhang.seckill.kafka;

import com.joizhang.seckill.api.kafka.SecKillMessage;
import com.joizhang.seckill.api.vo.SecKillGoodsVo;
import com.joizhang.seckill.api.vo.UserVo;
import com.joizhang.seckill.redis.RedisService;
import com.joizhang.seckill.service.GoodsServiceImpl;
import com.joizhang.seckill.service.OrderServiceImpl;
import com.joizhang.seckill.service.SecKillServiceImpl;
import com.joizhang.seckill.utils.BeanUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.Iterator;
import java.util.List;

@Component
public class MQReceiver {

    private static final Logger log = LoggerFactory.getLogger(MQReceiver.class);

    final RedisService redisService;

    final GoodsServiceImpl goodsService;

    final OrderServiceImpl orderService;

    final SecKillServiceImpl secKillService;

    public MQReceiver(RedisService redisService, GoodsServiceImpl goodsService, OrderServiceImpl orderService,
                      SecKillServiceImpl secKillService) {
        this.redisService = redisService;
        this.goodsService = goodsService;
        this.orderService = orderService;
        this.secKillService = secKillService;
    }

    @KafkaListener(topics = MQConfig.SECKILL_QUEUE, containerFactory = "ackContainerFactory")
    public void secKillReceive(List<ConsumerRecord<String, String>> consumerRecords, Acknowledgment ack) {
        log.info("receive batch: " + consumerRecords.size());
        for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
            String message = consumerRecord.value();
            if (StringUtils.isEmpty(message)) {
                continue;
            }
            SecKillMessage secKillMessage = BeanUtil.stringToBean(message, SecKillMessage.class);
            secKillService.handleSecKillMessage(secKillMessage);
        }
        // 提交offset
        ack.acknowledge();
    }

    @KafkaListener(topics = MQConfig.TEST_QUEUE)
    public void receive(String message) {
        log.info("receive queue message: " + message);
    }

}
